package com.yanggu.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.IOUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.util.Arrays;
import java.util.stream.Stream;

/**
 * @version V1.0
 * @author: YangGu
 * @date: 2019/12/7 21:43
 * @description: hdfs-client的demo演示
 */
public class HdfsClient {

    private FileSystem fileSystem;

    private Configuration conf;

    @Before
    public void before() throws Exception {
        System.out.println("程序开始执行");
        //1. 创建配置对象
        conf = new Configuration();
        //2. 创建uri对象
        URI uri = new URI("hdfs://hadoop102:9000");
        //3. 设置用户
        String user = "yanggu";
        //4. 得到fileSystem对象
        fileSystem = FileSystem.get(uri, conf, user);
    }

    @After
    public void after() throws Exception {
        //6. 关闭资源
        fileSystem.close();
        System.out.println("程序结束");
    }


    /**
     * 测试开发环境是否搭建成功
     * 创建文件夹
     *
     * @throws Exception
     */
    @Test
    public void testMkdir() throws Exception {
        //5. 通过fileSystem对象进行增删改查的操作
        boolean result = fileSystem.mkdirs(new Path("/user/yanggu"));
        System.out.println("创建文件夹的结果：" + result);
    }


    /**
     * hdsf的文件上传(并且测试参数的优先级)
     * 优先级从高到低
     * 程序中设置的参数 > classpath下的配置文件中的参数 > 服务器配置文件中的参数
     *
     * @throws Exception
     */
    @Test
    public void testCopyFromLocalFile() throws Exception {
        //5. 将本地的文件复制到hdfs中
        fileSystem.copyFromLocalFile(new Path("e:/banzhang.txt"), new Path("/user/yanggu/"));
    }


    /**
     * 将文件从hdfs中copy到local中
     *
     * @throws Exception
     */
    @Test
    public void testCopyToLocalFile() throws Exception {
        //boolean delSrc, Path src, Path dst, boolean useRawLocalFileSystem
        //1. 是否删除源数据
        //2. 源路径
        //3. 目标路径
        //4. 是否使用本地文件系统。false会产生crc文件。true不会产生crc文件
        //crc: 循环冗余校验（Cyclic Redundancy Check， CRC）是一种根据网络数据包或计算机文件等数据产生简短固定位数校验码的一种信道编码技术，主要用来检测或校验数据传输或者保存后可能出现的错误。它是利用除法及余数的原理来作错误侦测的。
        fileSystem.copyToLocalFile(false, new Path("/user/yanggu/banzhang.txt"), new Path("e:/banzhang2.txt"), true);
    }


    /**
     * 删除hdfs中的文件
     *
     * @throws Exception
     */
    @Test
    public void testDelete() throws Exception {
        boolean delete = fileSystem.delete(new Path("/user/yanggu"), true);
        System.out.println("是否删除成功" + delete);
    }


    /**
     * 更改文件的名称
     *
     * @throws Exception
     */
    @Test
    public void testRename() throws Exception {
        boolean rename = fileSystem.rename(new Path("/user/yanggu/banzhang.txt"), new Path("banhua.txt"));
        System.out.println("是否改名成功：" + rename);
    }


    /**
     * 查看文件名称、权限、长度、块信息等
     * -rw-r--r--	yanggu	supergroup	12 B	2019/12/7 下午11:06:38	1	128 MB	banhua.txt
     *
     * @throws Exception
     */
    @Test
    public void testListFiles() throws Exception {
        RemoteIterator<LocatedFileStatus> listFiles = fileSystem.listFiles(new Path("/"), true);
        while (listFiles.hasNext()) {
            LocatedFileStatus next = listFiles.next();
            //获得文件名
            String filename = next.getPath().getName();
            System.out.println("文件名是:" + filename);
            //获得拥有者
            String owner = next.getOwner();
            System.out.println("文件的拥有者:" + owner);
            //文件的组
            String group = next.getGroup();
            System.out.println("文件的组：" + group);
            //文件的权限
            FsPermission permission = next.getPermission();
            System.out.println("文件的权限" + permission);
            //文件的块信息
            Stream.of(next.getBlockLocations()).forEach(blockLocation -> {
                try {
                    //打印主机的信息
                    System.out.println(Arrays.toString(blockLocation.getHosts()));
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
        }
    }


    /**
     * HDFS文件和文件夹判断
     *
     * @throws Exception
     */
    @Test
    public void testListStatus() throws Exception {
        RemoteIterator<LocatedFileStatus> listFiles = fileSystem.listFiles(new Path("/"), true);
        while (listFiles.hasNext()) {
            LocatedFileStatus fileStatus = listFiles.next();
            // 如果是文件
            if (fileStatus.isFile()) {
                System.out.println("f:" + fileStatus.getPath().getName());
            } else {
                System.out.println("d:" + fileStatus.getPath().getName());
            }
        }
    }


    /**
     * 通过io流的方式将本地文件上传到hdfs中
     */
    @Test
    public void putFileToHdfs() throws Exception {
        //创建输入流
        FileInputStream fileInputStream = new FileInputStream(new File("e:/banzhang.txt"));
        //创建输出流
        FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("/user/yanggu/banzhang.txt"));
        //流中的字节的copy
        IOUtils.copyBytes(fileInputStream, fsDataOutputStream, conf);
        //关闭流
        IOUtils.closeStream(fsDataOutputStream);
        IOUtils.closeStream(fileInputStream);

    }


    /**
     * 从hdfs中copy到本地
     *
     * @throws Exception
     */
    @Test
    public void getFileFromHdfs() throws Exception {
        //创建输出流
        FileOutputStream fileOutputStream = new FileOutputStream(new File("e:/banhua.txt"));
        //获得输入流
        FSDataInputStream fsDataInputStream = fileSystem.open(new Path("/user/yanggu/banhua.txt"));
        //流的拷贝
        IOUtils.copyBytes(fsDataInputStream, fileOutputStream, conf);
        //关闭流
        IOUtils.closeStream(fsDataInputStream);
        //关闭流
        IOUtils.closeStream(fileOutputStream);
    }


    /**
     * 分块读取HDFS上的大文件，比如根目录下的/hadoop-2.7.2.tar.gz
     * 先下载第一块
     *
     * @throws Exception
     */
    @Test
    public void readFileSeek1() throws Exception {
        //创建输出流
        FileOutputStream fileOutputStream = new FileOutputStream(new File("e:/hadoop-2.7.2.tar.gz.part1"));
        //获得输入流
        FSDataInputStream fsDataInputStream = fileSystem.open(new Path("/hadoop-2.7.2.tar.gz"));
        //限制读取128m的大小
        byte[] buffer = new byte[1024];
        int length;
        for (int i = 0; i < 1024 * 128; i++) {
            while ((length = fsDataInputStream.read(buffer)) != -1) {
                fileOutputStream.write(buffer, 0, length);
            }
        }
        //关闭流
        IOUtils.closeStream(fsDataInputStream);
        //关闭流
        IOUtils.closeStream(fileOutputStream);
    }


    /**
     * 下载第二块
     *
     * @throws Exception
     */
    @Test
    public void readFileSeek2() throws Exception {
        FileOutputStream fileOutputStream = new FileOutputStream(new File("e:/hadoop-2.7.2.tar.gz.part2"));
        FSDataInputStream fsDataInputStream = fileSystem.open(new Path("/hadoop-2.7.2.tar.gz"));
        //定位输入数据的位置
        fsDataInputStream.seek(1024 * 1024 * 128);
        byte[] buffer = new byte[1024];
        int length;
        while ((length = fsDataInputStream.read(buffer)) != -1) {
            fileOutputStream.write(buffer, 0, length);
        }
        //关闭流
        IOUtils.closeStream(fsDataInputStream);
        //关闭流
        IOUtils.closeStream(fileOutputStream);
    }
}

